package org.rcy.framework.mq.serializer;

import com.fasterxml.jackson.databind.ObjectMapper;
import org.apache.kafka.common.errors.SerializationException;
import org.apache.kafka.common.header.Headers;
import org.rcy.framework.api.mq.AbstractMessage;
import org.springframework.kafka.support.JacksonUtils;
import org.springframework.kafka.support.converter.AbstractJavaTypeMapper;
import org.springframework.kafka.support.converter.DefaultJackson2JavaTypeMapper;
import org.springframework.kafka.support.converter.Jackson2JavaTypeMapper;
import org.springframework.lang.Nullable;
import org.springframework.util.Assert;
import org.springframework.util.ClassUtils;
import org.springframework.util.StringUtils;

import java.io.IOException;
import java.util.HashMap;
import java.util.Map;

/**
 * @author 吴康桥
 * @Description
 * @date 2022/11/30 16:51
 */


public class KafkaJsonSerializer <M extends AbstractMessage> implements org.apache.kafka.common.serialization.Serializer<M>{
	public static final String ADD_TYPE_INFO_HEADERS = "spring.json.add.type.headers";
	public static final String TYPE_MAPPINGS = "spring.json.type.mapping";
	protected final ObjectMapper objectMapper;
	protected boolean addTypeInfo;
	protected Jackson2JavaTypeMapper typeMapper;
	private boolean typeMapperExplicitlySet;

	public KafkaJsonSerializer() {
		this(JacksonUtils.enhancedObjectMapper());
	}

	public KafkaJsonSerializer(ObjectMapper objectMapper) {
		this.addTypeInfo = true;
		this.typeMapper = new DefaultJackson2JavaTypeMapper();
		this.typeMapperExplicitlySet = false;
		Assert.notNull(objectMapper, "'objectMapper' must not be null.");
		this.objectMapper = objectMapper;
	}

	public boolean isAddTypeInfo() {
		return this.addTypeInfo;
	}

	public void setAddTypeInfo(boolean addTypeInfo) {
		this.addTypeInfo = addTypeInfo;
	}

	public Jackson2JavaTypeMapper getTypeMapper() {
		return this.typeMapper;
	}

	public void setTypeMapper(Jackson2JavaTypeMapper typeMapper) {
		Assert.notNull(typeMapper, "'typeMapper' cannot be null");
		this.typeMapper = typeMapper;
		this.typeMapperExplicitlySet = true;
	}

	public void setUseTypeMapperForKey(boolean isKey) {
		if (!this.typeMapperExplicitlySet && this.getTypeMapper() instanceof AbstractJavaTypeMapper) {
			((AbstractJavaTypeMapper)this.getTypeMapper()).setUseForKey(isKey);
		}

	}

	public void configure(Map<String, ?> configs, boolean isKey) {
		this.setUseTypeMapperForKey(isKey);
		if (configs.containsKey("spring.json.add.type.headers")) {
			Object config = configs.get("spring.json.add.type.headers");
			if (config instanceof Boolean) {
				this.addTypeInfo = (Boolean)config;
			} else {
				if (!(config instanceof String)) {
					throw new IllegalStateException("spring.json.add.type.headers must be Boolean or String");
				}

				this.addTypeInfo = Boolean.valueOf((String)config);
			}
		}

		if (configs.containsKey("spring.json.type.mapping") && !this.typeMapperExplicitlySet && this.typeMapper instanceof AbstractJavaTypeMapper) {
			((AbstractJavaTypeMapper)this.typeMapper).setIdClassMapping(createMappings((String)configs.get("spring.json.type.mapping")));
		}

	}

	protected static Map<String, Class<?>> createMappings(String mappings) {
		Map<String, Class<?>> mappingsMap = new HashMap();
		String[] array = StringUtils.commaDelimitedListToStringArray(mappings);
		String[] var3 = array;
		int var4 = array.length;

		for(int var5 = 0; var5 < var4; ++var5) {
			String entry = var3[var5];
			String[] split = entry.split(":");
			Assert.isTrue(split.length == 2, "Each comma-delimited mapping entry must have exactly one ':'");

			try {
				mappingsMap.put(split[0].trim(), ClassUtils.forName(split[1].trim(), ClassUtils.getDefaultClassLoader()));
			} catch (LinkageError | ClassNotFoundException var9) {
				throw new IllegalArgumentException(var9);
			}
		}

		return mappingsMap;
	}

	@Nullable
	public byte[] serialize(String topic, Headers headers, @Nullable M data) {
		if (data == null) {
			return null;
		} else {
			if (this.addTypeInfo && headers != null) {
				this.typeMapper.fromJavaType(this.objectMapper.constructType(data.getClass()), headers);
			}

			return this.serialize(topic, data);
		}
	}

	@Nullable
	public byte[] serialize(String topic, @Nullable M data) {
		if (data == null) {
			return null;
		} else {
			try {
				return this.objectMapper.writeValueAsBytes(data);
			} catch (IOException var4) {
				throw new SerializationException("Can't serialize data [" + data + "] for topic [" + topic + "]", var4);
			}
		}
	}

	public void close() {
	}

	public KafkaJsonSerializer<M> forKeys() {
		this.setUseTypeMapperForKey(true);
		return this;
	}

	public KafkaJsonSerializer<M> noTypeInfo() {
		this.setAddTypeInfo(false);
		return this;
	}

	public KafkaJsonSerializer<M> typeMapper(Jackson2JavaTypeMapper mapper) {
		this.setTypeMapper(mapper);
		return this;
	}
}
